package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.common.utils.JDBCUtil.Conversions
import com.lvmama.rhino.common.utils.Utils
import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._


/**
  * 流量来向、去向
  * Created by yuanxiaofeng on 2016/10/22.
  */
object WirelessTrace {

  import Conversions._
  import com.lvmama.rhino.analyze.client.WirelessStat._

  def process(sc: SparkContext,  pageForward: DataFrame): Unit ={
    val temp = pageForward.filter(!col("buttonCode").equalTo("back")).coalesce(20)
      .withColumn("productId", when(col("pageParam").getItem("productId").isNull, "-1").otherwise(col("pageParam").getItem("productId")))
      .withColumn("categoryId", when(col("pageParam").getItem("categoryId").isNull, "-1").otherwise(col("pageParam").getItem("categoryId")))
      .withColumn("sk", when(col("pageParam").getItem("mt").equalTo("index"), col("pageParam").getItem("sk")).otherwise(""))
      .withColumn("pageTotalCode", concat(col("pageTypeCode") , col("categoryId") , col("productId")))
    val temp1 = temp.withColumn("parentCategoryId", dataLag(col("categoryId")))
      .withColumn("parentProductId", dataLag(col("productId")))
      .withColumn("parentPageTotalCode", dataLag(col("pageTotalCode")))
      .withColumn("parentPageCode", dataLag(col("pageTypeCode")))
      .withColumn("parentSK", dataLag(col("sk")))
    val conversion  = temp1.filter((col("pageTotalCode") !== col("parentPageTotalCode")) || col("parentPageTotalCode").isNull)
    val resultDataFrame = conversion.select(col("pageTypeCode"), col("parentPageCode"), col("platformCode"),col("productId"), col("categoryId"),col("parentSK"), col("parentCategoryId"),col("parentProductId"))
      .groupBy(col("pageTypeCode"), col("parentPageCode"), col("categoryId"),col("productId"),col("parentSK"), col("parentCategoryId"),col("parentProductId"),col("platformCode"))
      .agg(count(col("*")).as("count"))/*.transform(filterUsefulData)*/
      .coalesce(20)
    val yesterday =  Utils.getYesterday()
    resultDataFrame.select(col("pageTypeCode").as("page_code"), col("parentPageCode").as("parent_page_code"), col("platformCode").as("platform_code"),
      when(col("productId").equalTo("-1"), null).otherwise(col("productId")).as("product_id"),
      when(col("categoryId").equalTo("-1"), null).otherwise(col("categoryId")).as("category_id"),
      when(col("parentCategoryId").equalTo("-1"), null).otherwise(col("parentCategoryId")).as("parent_category_id"),
      when(col("parentProductId").equalTo("-1"), null).otherwise(col("parentProductId")).as("parent_product_id"),
      lit(yesterday).as("oper_date"), col("count").as("amount"),col("parentSK").as("sk"))
      .insertDF2MysqlDirect("flow_statistics_conversion")
  }

  /**
    * 目前在插入数据库时，只有满足下列情况才会插入数据库，在以后回全部放开
    * 跟团游15，当地游16，酒店套餐17 ，度假酒店1，其他 null
    */
  val filterUsefulData = (df: DataFrame) =>
    df.filter(col("parentCategoryId") === 1 || col("parentCategoryId") === 15 || col("parentCategoryId") === 16
      || col("parentCategoryId") === 17 || col("parentCategoryId") === -1 || col("parentCategoryId") === 11
      || col("parentCategoryId") === 12 || col("parentCategoryId") === 13 || col("parentCategoryId").isNull)
}
